package gupao.concurrency.kms;

import gupao.concurrency.utils.SigarUtil;
import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import gupao.concurrency.utils.FutureUtil.SuccFailCount;

import static gupao.concurrency.utils.FutureUtil.waitFuturesComplete;
import static gupao.concurrency.utils.FutureUtil.waitFuturesComplete2;
import static java.text.MessageFormat.format;

@Slf4j
public class ValidateMainTester {
    @RequiredArgsConstructor
    static class CallableTask implements
            Callable<SuccFailCount>{
        int succ = 0, fail = 0;
        final List<RequestModel> partition;
        @Override
        public SuccFailCount call() {
            for (RequestModel req : partition) {
                if (validator.validateRequest(req, "rf")) {
                    succ++;
                } else {
                    fail++;
                }
            }
            return new SuccFailCount(succ,fail);
        }
    }
    static final int concurrency = 1;
    static final int partitionNum = concurrency;
    static final int cpuNum = Runtime.getRuntime().availableProcessors();
    static SimpleRequestValidator validator = new SimpleRequestValidator();
    static ThreadPoolExecutor executor = ThreadPoolService
            .getOrCreateThreadPoolExecutor("validator", concurrency, concurrency);
    final static ConcurrentHashMap<Long, String> threadIdNameMap = new ConcurrentHashMap<>(512);

    public static void main(String[] args) {
        System.out.println(cpuNum);
        Utils.warmUpThreads(executor, threadIdNameMap, concurrency);
        List<RequestModel> mockRequests = RequestListMock.mockRequests();
        doValidateTest(mockRequests);
//        doValidateTest2(mockRequests);
    }

    static void doValidateTest2(List<RequestModel> requests){
        AtomicBoolean stop = new AtomicBoolean(false);
        AtomicInteger totalSucc = new AtomicInteger(0);
        AtomicInteger totalFail = new AtomicInteger(0);
        System.out.println("start");
        RequestModel p = new RequestModel(null,null);
        BlockingQueue<RequestModel> reqQueue = new LinkedBlockingQueue<>(requests);
        reqQueue.add(p);
        System.out.println("queue inited");
        List<Future<?>> fsList = new ArrayList<Future<?>>();
        long start = System.currentTimeMillis();
        long prvThreadTime = SigarUtil.calcThreadsTime(threadIdNameMap.keySet());
        long prvCpuTime = SigarUtil.calcCpuTime();
        for(int i = 0; i < concurrency; i++){
            Future<?> f = executor.submit(() -> {
                while(!stop.get()){
                    try {
                        RequestModel req = reqQueue.poll(1, TimeUnit.NANOSECONDS);
                        if(req == p){
                            stop.set(true);
                        } else {
                            if(req != null){
                                if (validator.validateRequest(req, "rf")) {
                                    totalSucc.incrementAndGet();
                                } else {
                                    totalFail.incrementAndGet();
                                }
                            }
                        }
                    } catch (InterruptedException e) {
                        log.error("",e);
                    }
                }
            });
            fsList.add(f);
        }
        waitFuturesComplete(fsList);
        long curThreadTime = SigarUtil.calcThreadsTime(threadIdNameMap.keySet());
        long curCpuTime = SigarUtil.calcCpuTime();
        long threadsTimeCost = (curThreadTime - prvThreadTime)/1000000;
        long cpuTimeCost = curCpuTime - prvCpuTime;
        System.out.println(format("Completed with succ: {0}, fail: {1}. time cost: {2}ms",
                totalSucc.get(), totalFail.get(), (System.currentTimeMillis() - start)));

        System.out.println("Threads time: " + threadsTimeCost);
        System.out.println("CPU time: " + cpuTimeCost);
        double ratio = threadsTimeCost * 1.0D / cpuTimeCost;
        System.out.println("CPU Ratio: " + ratio);
    }

    private static void doValidateTest(List<RequestModel> requests) {
        long start = System.currentTimeMillis();
//        List<List<RequestModel>> partitions = split(requests, partitionNum);
        List<List<RequestModel>> partitions = split2(requests, partitionNum);
        printPartitions(partitions, concurrency, partitionNum);
        System.out.println(format("split partitions time cost: {0}ms, start running...",
                (System.currentTimeMillis() - start)));
        start = System.currentTimeMillis();
        AtomicInteger totalSucc = new AtomicInteger(0);
        AtomicInteger totalFail = new AtomicInteger(0);
//        List<Future<?>> fsList = partitions.stream().map(
//                partition -> executor.submit(() -> {
//                        for (RequestModel req : partition) {
//                            if (validator.validateRequest(req, "rf")) {
//                                totalSucc.incrementAndGet();
//                            } else {
//                                totalFail.incrementAndGet();
//                            }
//                    }
//                })).collect(Collectors.toList());
//
//        long prvThreadTime = SigarUtil.calcThreadsTime(threadIdNameMap.keySet());
//        long prvCpuTime = SigarUtil.calcCpuTime();
//        waitFuturesComplete(fsList);
        long prvThreadTime = SigarUtil.calcThreadsTime(threadIdNameMap.keySet());
        long prvCpuTime = SigarUtil.calcCpuTime();
        List<Future<SuccFailCount>> fsList = partitions.stream().map(
                partition -> executor.submit(new CallableTask(partition))).collect(Collectors.toList());
        List<SuccFailCount> succFailCounts = waitFuturesComplete2(fsList);
        long curThreadTime = SigarUtil.calcThreadsTime(threadIdNameMap.keySet());
        long curCpuTime = SigarUtil.calcCpuTime();
        long threadsTimeCost = (curThreadTime - prvThreadTime)/1000000;
        long cpuTimeCost = curCpuTime - prvCpuTime;
        succFailCounts.forEach(c -> {
            totalSucc.addAndGet(c.getSucc());
            totalFail.addAndGet(c.getFail());
        });
        System.out.println(format("Completed with succ: {0}, fail: {1}. time cost: {2}ms",
                totalSucc.get(), totalFail.get(), (System.currentTimeMillis() - start)));
        System.out.println("Threads time: " + threadsTimeCost);
        System.out.println("CPU time: " + cpuTimeCost);
        double ratio = threadsTimeCost * 1.0D / cpuTimeCost;
        System.out.println("CPU Ratio: " + ratio);
    }

    private static void printPartitions(List<List<RequestModel>> partitions, int concurrency, int partitionNum) {
        int total = 0;
        for (int i = 0; i < partitions.size(); i++) {
            System.out.println(format("Request Partition {0} size: {1}", i, partitions.get(i).size()));
            total += partitions.get(i).size();
        }
        System.out.println(format("concurrency: {0}, partition Number: {1}", concurrency, partitions.size()));
        System.out.println("Total Requests: " + total);
    }

    public static List<List<RequestModel>> split2(
            List<RequestModel> requests, int partitionNum){
        List<List<RequestModel>> partitions = new ArrayList<>(partitionNum);
        IntStream.range(0, partitionNum)
                .forEach(i -> partitions.add(new ArrayList<>()));
        int i = 0;
        for (RequestModel request : requests) {
            int index = i++ % partitionNum;
            partitions.get(index).add(request);
        }
        return partitions;
    }

    private static List<List<RequestModel>> split(
            List<RequestModel> requests, int partition) {
        List<List<RequestModel>> partitions = new ArrayList<>(partition);
        IntStream.range(0, partition)
                .forEach(i -> partitions.add(new ArrayList<>()));
        //分配每个 partition 对应的 request
        int num = 0;
        int reqNumPerPartition = requests.size() / partition;
        for (int idx = 0; idx < partition; idx++) {
            List<RequestModel> partitionIdx = partitions.get(idx);
            int start = idx * reqNumPerPartition;
            int end = start + reqNumPerPartition;
            for (num = start; num < end; num++) {
                partitionIdx.add(requests.get(num));
            }
        }
        if (num < requests.size()) {
            List<RequestModel> last = new ArrayList<>();
            partitions.add(last);
            for (int i = num; i < requests.size(); i++) {
                last.add(requests.get(i));
            }
        }
        return partitions;
    }

}
